package com.csw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;

public class Demo02API {
    private Connection conn;
    private Admin admin;
    @Before
    public void init()throws IOException{
        //创建配置
        Configuration conf = HBaseConfiguration.create();
        //设置hbase  zookeeper连接地址
        conf.set("hbase.zookeeper.quorum","master:2181,node1:2181,node2:2181");
        //创建与hbase的连接
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }
    @Test
    public void create_table()  throws IOException{
        //创建表
        HTableDescriptor table1 = new HTableDescriptor(TableName.valueOf("table1"));
        //设置列族
        HColumnDescriptor csw1 = new HColumnDescriptor("csw1");
        //设置VERSIONS
        csw1.setVersions(1,5);
        table1.addFamily(csw1);

        admin.createTable(table1);
    }
    @Test
    public void drop_table()throws Exception{
        //要删除表先停止表的运行将表变为disable状态
        admin.disableTable(TableName.valueOf("table1"));
        //删除表
        admin.deleteTable(TableName.valueOf("table1"));
    }
    @Test
    public void put()throws Exception{
        //先获取表
        Table user = conn.getTable(TableName.valueOf("User"));
        //put一个rowkey
        Put put = new Put("row777".getBytes());
        //传入列族名、列名、和value(要传入的数据)
        put.addColumn("info".getBytes(),"name".getBytes(),"zhangsan".getBytes());
        user.put(put);
    }
    @Test
    public void get()throws Exception{
        //先获取表
        Table user = conn.getTable(TableName.valueOf("User"));

        //指定rowkey
        Get get = new Get("row777".getBytes());
        Result rs = user.get(get);
        //获取rowkey
        byte[] row = rs.getRow();
        System.out.println(Bytes.toString(row));
        //指定列族和列名去获取数据
        byte[] value = rs.getValue("info".getBytes(), "name".getBytes());
        System.out.println(Bytes.toString(value));
    }
    @Test
    public void scan()throws Exception{
        Table user = conn.getTable(TableName.valueOf("User"));
        //创建一个Scan对象
        Scan scan = new Scan();
        //设置scan的startRow，表示从哪个rowkey开始，包含本身
        scan.withStartRow("row1".getBytes());
        //设置scan的stopRow，表示从哪个rowkey结束，不包含本身
        scan.withStopRow("row4".getBytes());
        //使用getScanner方法，scan   table
        ResultScanner scanner = user.getScanner(scan);
        Result rs = scanner.next();
        while (rs!=null){
            //获取rowkey
            byte[] row = rs.getRow();
            //把rowkey的Bytes格式转为String
            String rowkey = Bytes.toString(row);
            String colume="";
            //把获取到的rowkey和列族进行比较，指定每个列族里的列名从而获取value
            if (rowkey.equals("row1")){
                colume="name";
            }else if (rowkey.equals("row1111")){
                colume="name";
            }else if (rowkey.equals("row2")){
                colume="age";
            }else if (rowkey.equals("row3")){
                colume="sex";
            }
            //获取value
            byte[] value = rs.getValue("info".getBytes(), colume.getBytes());
            System.out.println(Bytes.toString(value));

            rs = scanner.next();
        }
    }
    @Test
    public void putAll()throws Exception{
        //判断表不存在就创建
        if (!admin.tableExists(TableName.valueOf("students"))){
            //设置表名
            HTableDescriptor students = new HTableDescriptor(TableName.valueOf("students"));
            //设置表的列族
            HColumnDescriptor info = new HColumnDescriptor("info");
//            设置最大版本
//            info.setMaxVersions(5);
            students.addFamily(info);
            admin.createTable(students);
        }
        //BufferedReader读取数据
        BufferedReader br = new BufferedReader(new FileReader("data/students.txt"));
        String line=br.readLine();
        //获取表
        Table students = conn.getTable(TableName.valueOf("students"));
        while (line!=null){
            //获取数据
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            int age = Integer.valueOf(split[2]);
            String gender = split[3];
            String clazz = split[4];
            //以id作为rowkey
            Put put = new Put(id.getBytes());
            put.addColumn("info".getBytes(),"name".getBytes(),name.getBytes());
            // int 类型的数据无法使用getBytes方法
            put.addColumn("info".getBytes(), "age".getBytes(), Bytes.toBytes(age));
            put.addColumn("info".getBytes(),"gender".getBytes(),gender.getBytes());
            put.addColumn("info".getBytes(),"clazz".getBytes(),clazz.getBytes());
            students.put(put);
            // 读取下一条数据
            line=br.readLine();
        }
    }
    @Test
    public void cells()throws Exception{
        //获取表名
        Table students = conn.getTable(TableName.valueOf("students".getBytes()));
        Scan scan = new Scan();
        //设置rowkey范围
        scan.withStartRow("1500100001".getBytes());
        scan.withStopRow("1500100021".getBytes());
        //使用getScanner方法，scan   table
        ResultScanner scanner = students.getScanner(scan);
        for (Result rs : scanner){
            //获取rowkey
            byte[] row = rs.getRow();
            //转化rowkey格式为String
            String rowkey = Bytes.toString(row);
//             直接通过rs.getValue可能会拿不到数据，因为每条数据它的列不一样
//            rs.getValue("info".getBytes(), "name".getBytes());
//            rs.getValue("info".getBytes(), "age".getBytes());
//            rs.getValue("info".getBytes(), "gender".getBytes());
//            rs.getValue("info".getBytes(), "clazz".getBytes());
            System.out.println("当前行的RowKey为："+rowkey);

            List<Cell> cells = rs.listCells();
            for (Cell cell : cells) {
//                 获取rowkey
//                String row = Bytes.toString(CellUtil.cloneRow(cell));
                //获取列族
                byte[] bytes = CellUtil.cloneFamily(cell);
                String family = Bytes.toString(bytes);
                //获取列
                byte[] bytes1 = CellUtil.cloneQualifier(cell);
                String qualifier = Bytes.toString(bytes1);
                //获取cell数据
                byte[] valuebytes = CellUtil.cloneValue(cell);
                String value;
                if ("age".equals(qualifier)){
                    value=String.valueOf(Bytes.toInt(valuebytes));
                }else{
                    value=Bytes.toString(valuebytes);
                }
                System.out.println("当前获取的列族名为："+family+",当前获取的列名为:"+qualifier+",cell值为:"+value);
            }
        }
    }
    @After
    public void close() throws IOException {
        conn.close();
    }
}
